package cn.ac.iscas.service.source;

import cn.ac.iscas.base.utils.*;
import cn.ac.iscas.config.MiddlePlatformConfig;
import cn.ac.iscas.domain.response.middleplatform.MetaColumnDTO;
import cn.ac.iscas.domain.response.middleplatform.MetaTableDTO;
import cn.ac.iscas.domain.response.middleplatform.ResponseEntity;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;
import com.iscas.datasong.lib.common.DataSongException;
import com.iscas.datasong.lib.util.DataSongJsonUtils;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static cn.ac.iscas.base.constant.JdbcConnectorConstant.*;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/5/30 16:08
 */
@Service
public class MySqlSource {
//    @Autowired
//    private MiddlePlatformConfig middlePlatformConfig;
    private final static String ID = "mysql";
    private final static String CONNECTOR_DRIVER_VALUE = "com.mysql.jdbc.Driver";
    private final static String CONNECTOR_TYPE_VALUE = "jdbc";
    private OkHttpClientUtils httpClient = new OkHttpClientUtils();
    private String jobId;
    private String componentId;
    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public MySqlSource(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public MySqlSource() {

    }


    /**
     * @description:
     * CREATE TABLE category_dim (
     *     sub_category_id BIGINT, dash
     *     parent_category_id BIGINT
     * ) WITH (
     *     'connector' = 'jdbc',
     *     'url' = 'jdbc:mysql://localhost:3306/flink',qwe123qwe123
     *
     *     'table-name' = 'category',
     *     'driver' = 'com.mysql.jdbc.Driver',
     *     'username' = 'root',
     *     'password' = '123456',
     *     'lookup.cache.max-rows' = '5000',
     *     'lookup.cache.ttl' = '10min'
     * );
     * @param: parameters
     * @return: java.lang.String
     * @author LJian
     * @date: 2022/6/6 10:51
     */
//    public String getSql(Map<String, Object> parameters) {
//        String ip = ParameterUtils.getRequiredString(parameters, "ip");
//        String port = ParameterUtils.getRequiredString(parameters, "port");
//        String db = ParameterUtils.getRequiredString(parameters, "database");
//        String table = ParameterUtils.getRequiredString(parameters, "table");
//        table = String.format("%s.%s.%s", jobId, jobId, table);
//        this.currentTableName = table;
//        String datasource = ParameterUtils.getRequiredString(parameters, "datasource");
//        String username = "";
//        String password = "";
//        try {
//            username = AesUtils.aesDecrypt(ParameterUtils.getRequiredString(parameters, "username"));
//            password = AesUtils.aesDecrypt(ParameterUtils.getRequiredString(parameters, "password"));
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//
//        String jdbcUrl = new StringBuffer("jdbc:mysql://").append(ip).append(":").append(port).append("/").append(db).toString().trim();
//        StringBuffer withBuf = new StringBuffer("WITH (")
//                .append(String.format("'%s' = '%s'", CONNECTOR_TYPE_KEY, CONNECTOR_TYPE_VALUE))
//                .append(String.format(",'%s' = '%s'", CONNECTOR_URL_KEY, jdbcUrl))
//                .append(String.format(",'%s' = '%s'", CONNECTOR_TABLE_KEY, table))
//                .append(String.format(",'%s' = '%s'", CONNECTOR_USERNAME_KEY, username))
//                .append(String.format(",'%s' = '%s'", CONNECTOR_PASSWORD_KEY, password));
//
//        String withStr = withBuf.append(")").toString();
//        StringBuffer tableSchemaBuf = new StringBuffer(String.format("CREATE TABLE %s (", table));
//
//        MiddlePlatformConfig middlePlatformConfig = SpringUtil.getBean(MiddlePlatformConfig.class);
//        String tableMetaUrl = middlePlatformConfig.getQueryDataSourceTablemetaUrl() + datasource;
//        ResponseEntity<MetaTableDTO> response = null;
//        try {
//
//            String tableMetaStr = httpClient.doGet(tableMetaUrl);
//            response = DataSongJsonUtils.fromJson(tableMetaStr, ResponseEntity.class);
//        } catch (IOException e) {
//            e.printStackTrace();
//        } catch (DataSongException e) {
//            e.printStackTrace();
//        }
//        List<Map<String, Object>> metaTables = (List<Map<String, Object>>) response.getValue();
//        List<String> primaryKeys = new ArrayList<>();
//        List<FlinkSourceColumn> columns = new ArrayList<>();
//        for (Map<String, Object> metaTable : metaTables) {
//            if (metaTable.getOrDefault("tableName", "").toString().equalsIgnoreCase(table)) {
//                MetaTableDTO metaTableDTO = null;
//                try {
//                    metaTableDTO = DataSongJsonUtils.fromJson(metaTable.getOrDefault("tableMeta", "{}").toString(), MetaTableDTO.class);
//                } catch (DataSongException e) {
//                    e.printStackTrace();
//                }
//
//                for (MetaColumnDTO column : metaTableDTO.getMetaColumns()) {
//                    tableSchemaBuf.append(String.format("%s %s, ", column.getName(), TypeConvertUtil.mysqlTypeToFlinkTypeString(column.getColumnType())));
//                    if (column.isPrimary()) {
//                        primaryKeys.add(column.getName());
//                    }
//
//                    FlinkSourceColumn flinkSourceColumn = new FlinkSourceColumn();
//                    flinkSourceColumn.setName(column.getName());
//                    flinkSourceColumn.setType(TypeConvertUtil.mysqlTypeToFlinkTypeString(column.getColumnType()));
//                    flinkSourceColumn.setPrimaryKey(column.isPrimary());
//                    flinkSourceColumn.setDefaultVal(column.getDefaultVal());
//                    columns.add(flinkSourceColumn);
//                }
//            }
//        }
//
//        this.currentColumns = columns;
//
//
//        if (!primaryKeys.isEmpty()) {
//            tableSchemaBuf.append(String.format(" PRIMARY KEY (%s) NOT ENFORCED ", String.join(",", primaryKeys)));
//        }
//
//        String tableSchemaString = tableSchemaBuf.toString();
//        if (tableSchemaString.trim().endsWith(",")) {
//            tableSchemaString = tableSchemaString.substring(0, tableSchemaString.lastIndexOf(","));
//        }
//
//        tableSchemaString += ") ";
//
//        String createTableSql = tableSchemaString + withStr;
////        TableQueueService tableQueueService = SpringUtil.getBean(TableQueueService.class);
////        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8) {{ put(this.currentTableName, columns); }};
////        tableQueueService.put(this.jobId, this.componentId, tableColumns);
//        updateSource();
//        return createTableSql;
//    }

    public String getSql(Map<String, Object> parameters) {
        String host = ParameterUtils.getRequiredString(parameters, "host");
        String port = String.valueOf(ParameterUtils.getRequiredInt(parameters, "port"));
        String db = ParameterUtils.getRequiredString(parameters, "databaseName");
        String table = ParameterUtils.getRequiredString(parameters, "table");
        String tableWithCatalog = String.format("%s.%s.%s", jobId, jobId, table);
        this.currentTableName = tableWithCatalog;
        String datasourceName = ParameterUtils.getRequiredString(parameters, "datasourceName");
        String username = "";
        String password = "";
        try {
            username = AesUtils.aesDecrypt(ParameterUtils.getRequiredString(parameters, "username"));
            password = AesUtils.aesDecrypt(ParameterUtils.getRequiredString(parameters, "password"));
        } catch (Exception e) {
            e.printStackTrace();
        }

        String jdbcUrl = new StringBuffer("jdbc:mysql://").append(host).append(":").append(port).append("/").append(db).toString().trim()+ "?useUnicode=true&characterEncoding=UTF-8";
        StringBuffer withBuf = new StringBuffer("WITH (")
                .append(String.format("'%s' = '%s'", CONNECTOR_TYPE_KEY, CONNECTOR_TYPE_VALUE))
                .append(String.format(",'%s' = '%s'", CONNECTOR_URL_KEY, jdbcUrl))
                .append(String.format(",'%s' = '%s'", CONNECTOR_TABLE_KEY, table))
                .append(String.format(",'%s' = '%s'", CONNECTOR_USERNAME_KEY, username))
                .append(String.format(",'%s' = '%s'", CONNECTOR_PASSWORD_KEY, password));

        String withStr = withBuf.append(")").toString();
        StringBuffer tableSchemaBuf = new StringBuffer(String.format("CREATE TABLE %s (", tableWithCatalog));

        MiddlePlatformConfig middlePlatformConfig = SpringUtil.getBean(MiddlePlatformConfig.class);
        String tableMetaUrl = middlePlatformConfig.getQueryDataSourceTablemetaUrl() + datasourceName;
        ResponseEntity<MetaTableDTO> response = null;
        try {

            String tableMetaStr = httpClient.doGet(tableMetaUrl);
            response = DataSongJsonUtils.fromJson(tableMetaStr, ResponseEntity.class);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (DataSongException e) {
            e.printStackTrace();
        }
        List<Map<String, Object>> metaTables = (List<Map<String, Object>>) response.getValue();
        List<String> primaryKeys = new ArrayList<>();
        List<FlinkSourceColumn> columns = new ArrayList<>();
        for (Map<String, Object> metaTable : metaTables) {
            if (metaTable.getOrDefault("tableName", "").toString().equalsIgnoreCase(table)) {
                MetaTableDTO metaTableDTO = null;
                try {
                    metaTableDTO = DataSongJsonUtils.fromJson(metaTable.getOrDefault("tableMeta", "{}").toString(), MetaTableDTO.class);
                } catch (DataSongException e) {
                    e.printStackTrace();
                }

                for (MetaColumnDTO column : metaTableDTO.getMetaColumns()) {
                    tableSchemaBuf.append(String.format("%s %s, ", column.getName(), TypeConvertUtil.mysqlTypeToFlinkTypeString(column.getColumnType())));
                    if (column.isPrimary()) {
                        primaryKeys.add(column.getName());
                    }

                    FlinkSourceColumn flinkSourceColumn = new FlinkSourceColumn();
                    flinkSourceColumn.setName(column.getName());
                    flinkSourceColumn.setType(TypeConvertUtil.mysqlTypeToFlinkTypeString(column.getColumnType()));
                    flinkSourceColumn.setPrimaryKey(column.isPrimary());
                    flinkSourceColumn.setDefaultVal(column.getDefaultVal());
                    columns.add(flinkSourceColumn);
                }
            }
        }

        this.currentColumns = columns;


        if (!primaryKeys.isEmpty()) {
            tableSchemaBuf.append(String.format(" PRIMARY KEY (%s) NOT ENFORCED ", String.join(",", primaryKeys)));
        }

        String tableSchemaString = tableSchemaBuf.toString();
        if (tableSchemaString.trim().endsWith(",")) {
            tableSchemaString = tableSchemaString.substring(0, tableSchemaString.lastIndexOf(","));
        }

        tableSchemaString += ") ";

        String createTableSql = tableSchemaString + withStr;
//        TableQueueService tableQueueService = SpringUtil.getBean(TableQueueService.class);
//        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8) {{ put(this.currentTableName, columns); }};
//        tableQueueService.put(this.jobId, this.componentId, tableColumns);
        updateSource();
        return createTableSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }

    public static void main(String[] args) {
        Map<String, Object> params = new HashMap<>(32);
        params.put("ip", "172.16.10.159");
        params.put("port", "3306");
        params.put("db", "db_test_01");
        params.put("table", "ods_test_01");
        params.put("username", "2EPY14iZK5RDFsmzEdffJQ==");//root
        params.put("password", "2EPY14iZK5RDFsmzEdffJQ==");
        params.put("datasource", "db_test_01");

//        MySqlSource mySqlSource = new MySqlSource();
//        mySqlSource.getSql(params);


    }
}

